时会|可能会_Java网络编程——NIO处理写事件(SelectionKey.OP_WRITE)

篇首语：本文由编程笔记#小编为大家整理，主要介绍了Java网络编程——NIO处理写事件(SelectionKey.OP_WRITE)相关的知识，希望对你有一定的参考价值。

在前面NIO的例子中&＃xff0c;在服务端&＃xff0c;有对连接事件&＃xff08;SelectionKey.OP_ACCEPT&＃xff09;的处理&＃xff0c;也有对读事件&＃xff08;SelectionKey.OP_READ&＃xff09;的处理&＃xff0c;但是没有对写事件&＃xff08;SelectionKey.OP_WRITE&＃xff09;进行处理&＃xff0c;原因就是写事件有些特殊&＃xff0c;在这里单独记录一下。

网上有一些例子都是在服务端读完数据后直接给客户端SocketChannel对应的SelectionKey注册上写事件&＃xff08;SelectionKey.OP_WRITE&＃xff09;&＃xff0c;写完数据后也不注销写事件&＃xff1a;

&＃64;Slf4j
public class Nioserver
public static void main(String[] args) throws Exception
ServerSocketChannel serverSocketChannel &＃61; ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
serverSocketChannel.bind(new InetSocketAddress(8080));
Selector selector &＃61; Selector.open();
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
while (true)
int select &＃61; selector.select();
Set<SelectionKey> selectionKeys &＃61; selector.selectedKeys();
Iterator<SelectionKey> iterator &＃61; selectionKeys.iterator();
while (iterator.hasNext())
SelectionKey selectionKey &＃61; iterator.next();
if (selectionKey.isAcceptable())
ServerSocketChannel serverSocket &＃61; (ServerSocketChannel) selectionKey.channel();
SocketChannel socketChannel &＃61; serverSocket.accept();
log.info("receive connection from client. client:", socketChannel.getRemoteAddress());
socketChannel.configureBlocking(false);
socketChannel.register(selector, SelectionKey.OP_READ);
else if (selectionKey.isReadable())
ByteBuffer byteBuffer &＃61; ByteBuffer.allocate(1024);
SocketChannel socketChannel &＃61; (SocketChannel) selectionKey.channel();
socketChannel.read(byteBuffer);
String message &＃61; new String(byteBuffer.array()).trim();
byteBuffer.clear();
log.info("receive message from client. client: message length:", socketChannel.getRemoteAddress(),message.length());
socketChannel.register(selectionKey.selector(), SelectionKey.OP_WRITE);
selectionKey.attach(message);
else if (selectionKey.isWritable())
SocketChannel socketChannel &＃61; (SocketChannel) selectionKey.channel();
String response &＃61; (String) selectionKey.attachment();
ByteBuffer byteBuffer &＃61; ByteBuffer.allocate(response.length());
byteBuffer.put(response.getBytes());
byteBuffer.flip();
socketChannel.write(byteBuffer);
log.info("send message to client. client: message length:", socketChannel.getRemoteAddress(), response.length());

iterator.remove();






但实际上根本不能这么用&＃xff0c;即使在服务端向客户端写完数据后&＃xff0c;也会不断触发写事件&＃xff08;selector.select()返回&＃xff0c; selectionKey.isWritable()返回true &＃xff09;。

什么时候会触发写事件呢&＃xff1f;

在前面的文章 《Java网络编程——NIO的阻塞IO模式、非阻塞IO模式、IO多路复用模式的使用》 中简单提到SelectionKey.OP_WRITE事件表示已经可以向通道写数据了&＃xff08;通道目前可以用于写操作&＃xff09;&＃xff0c;那什么时候才算“可以向通道写数据”呢&＃xff1f;

如果有channel在Selector上注册了SelectionKey.OP_WRITE事件&＃xff0c;在调用selector.select();时&＃xff0c;系统会检查内核写缓冲区是否可写&＃xff08;当写缓冲区已满、channel调用了shutdownOutPut等情况&＃xff0c;内核缓冲区不可写&＃xff09;&＃xff0c;如果可写&＃xff0c;selector.select();会立即返回写事件。

把服务端处理读事件的代码优化如下&＃xff1a;

&＃64;Slf4j
public class NIOSelectorNonblockingWriteServer
private final static int MESSAGE_LENGTH &＃61; 1024 * 1024 * 100;
public static void main(String[] args) throws Exception
ServerSocketChannel serverSocketChannel &＃61; ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
serverSocketChannel.bind(new InetSocketAddress("127.0.0.1", 8080), 50);
Selector selector &＃61; Selector.open();
SelectionKey serverSocketKey &＃61; serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
while (true)
int count &＃61; selector.select();
log.info("select event count:" &＃43; count);
Set<SelectionKey> selectionKeys &＃61; selector.selectedKeys();
Iterator<SelectionKey> iterator &＃61; selectionKeys.iterator();
while (iterator.hasNext())
SelectionKey selectionKey &＃61; iterator.next();
// 有客户端请求建立连接
if (selectionKey.isAcceptable())
handleAccept(selectionKey);

// 有客户端发送数据
else if (selectionKey.isReadable())
handleRead(selectionKey);

// 可以向客户端发送数据
else if (selectionKey.isWritable())
handleWrite(selectionKey);

iterator.remove();



private static void handleAccept(SelectionKey selectionKey) throws IOException
ServerSocketChannel serverSocketChannel &＃61; (ServerSocketChannel) selectionKey.channel();
SocketChannel socketChannel &＃61; serverSocketChannel.accept();
if (Objects.nonNull(socketChannel))
log.info("receive connection from client. client:", socketChannel.getRemoteAddress());
// 设置客户端Channel为非阻塞模式&＃xff0c;否则在执行socketChannel.read()时会阻塞
socketChannel.configureBlocking(false);
Selector selector &＃61; selectionKey.selector();
socketChannel.register(selector, SelectionKey.OP_READ);


private static void handleRead(SelectionKey selectionKey) throws IOException
SocketChannel socketChannel &＃61; (SocketChannel) selectionKey.channel();
ByteBuffer readBuffer &＃61; ByteBuffer.allocate(MESSAGE_LENGTH);
int length &＃61; 0;
while (length < MESSAGE_LENGTH)
length &＃43;&＃61; socketChannel.read(readBuffer);

log.info("receive message from client. client: message length:", socketChannel.getRemoteAddress(), readBuffer.position());
ByteBuffer writeBuffer &＃61; ByteBuffer.allocate(readBuffer.position());
readBuffer.flip();
writeBuffer.put(readBuffer);
// 读完数据后&＃xff0c;为 SelectionKey 注册可写事件
if (!isInterest(selectionKey, SelectionKey.OP_WRITE))
selectionKey.interestOps(selectionKey.interestOps() &＃43; SelectionKey.OP_WRITE);

writeBuffer.flip();
selectionKey.attach(writeBuffer);

// 服务端可能是为每个Channel维护一块缓冲区&＃xff0c;当向某个Channel写数据时缓冲区满了&＃xff0c;还可以向其他Channel写数据
private static void handleWrite(SelectionKey selectionKey) throws IOException
SocketChannel socketChannel &＃61; (SocketChannel) selectionKey.channel();
ByteBuffer writeBuffer &＃61; (ByteBuffer) selectionKey.attachment();
int writeLength &＃61; socketChannel.write(writeBuffer);
log.info("send message to client. client: message length:", socketChannel.getRemoteAddress(), writeLength);
if (!writeBuffer.hasRemaining())
// 写完数据后&＃xff0c;要把写事件取消&＃xff0c;否则当写缓冲区有剩余空间时&＃xff0c;会一直触发写事件
selectionKey.interestOps(selectionKey.interestOps() - SelectionKey.OP_WRITE);
// socketChannel.shutdownOutput(); // channel调用shutdownOutput()后&＃xff0c;会停止触发写事件


// 判断 SelectionKey 对某个事件是否感兴趣
private static boolean isInterest(SelectionKey selectionKey, int event)
int interestSet &＃61; selectionKey.interestOps();
boolean isInterest &＃61; (interestSet & event) &＃61;&＃61; event;
return isInterest;





在服务端读取完客户端消息后&＃xff0c;会先判断当前SelectionKey是否已经注册了写事件&＃xff0c;如果没有则为其注册写事件。在服务端向客户端写完数据后&＃xff0c;会取消写事件。

在上面判断SelectionKey是否已经注册了某个事件时&＃xff0c;判断条件是 selectionKey.interestOps() & SelectionKey.OP_WRITE &＃xff0c;selectionKey.interestOps()就是已经注册的事件&＃xff0c;SelectionKey中可以只用1个整形数字来表示多个注册的事件&＃xff08;interestOps变量&＃xff09;&＃xff0c;SelectionKey.OP_READ&＃61;1&＃xff08;二进制为 00000001&＃xff09;&＃xff0c;SelectionKey.OP_WRITE&＃61;4&＃xff08;二进制为 00000100&＃xff09;&＃xff0c;SelectionKey.OP_CONNECT&＃61;8&＃xff08;二进制为 00001000&＃xff09;&＃xff0c;SelectionKey.OP_ACCEPT&＃61;16&＃xff08;二进制为 00010000&＃xff09;。当注册某个事件时&＃xff0c;会把对应事件对应的整数&＃xff08;单个事件对应的整数或者多个事件对应的整数和&＃xff09;赋值给interestOps变量&＃xff0c;比如注册读事件时&＃xff0c;interestOps&＃61;1&＃xff1b;注册读事件&＃43;写事件时&＃xff0c;interestOps&＃61;1&＃43;4 。因此可以通过按位与的算法来判断当前SelectionKey是否注册过某个事件。

为什么不在处理完读事件&＃xff08;selectionKey.isReadable()&＃xff09;后&＃xff0c;直接写数据&＃xff0c;而要重新注册一个读事件呢&＃xff1f;

那我们就继续看这个例子&＃xff1a;

&＃64;Slf4j
public class NIOSelectorBlockingWriteServer
private final static int MESSAGE_LENGTH &＃61; 1024 * 1024 * 100;
public static void main(String[] args) throws Exception
ServerSocketChannel serverSocketChannel &＃61; ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
serverSocketChannel.bind(new InetSocketAddress("127.0.0.1", 8080), 50);
Selector selector &＃61; Selector.open();
SelectionKey serverSocketKey &＃61; serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
while (true)
int count &＃61; selector.select();
log.info("select event count:" &＃43; count);
Set<SelectionKey> selectionKeys &＃61; selector.selectedKeys();
Iterator<SelectionKey> iterator &＃61; selectionKeys.iterator();
while (iterator.hasNext())
SelectionKey selectionKey &＃61; iterator.next();
// 有客户端请求建立连接
if (selectionKey.isAcceptable())
handleAccept(selectionKey);

// 有客户端发送数据
else if (selectionKey.isReadable())
handleRead(selectionKey);

iterator.remove();



private static void handleAccept(SelectionKey selectionKey) throws IOException
ServerSocketChannel serverSocketChannel &＃61; (ServerSocketChannel) selectionKey.channel();
SocketChannel socketChannel &＃61; serverSocketChannel.accept();
if (Objects.nonNull(socketChannel))
log.info("receive connection from client. client:", socketChannel.getRemoteAddress());
// 设置客户端Channel为非阻塞模式&＃xff0c;否则在执行socketChannel.read()时会阻塞
socketChannel.configureBlocking(false);
Selector selector &＃61; selectionKey.selector();
socketChannel.register(selector, SelectionKey.OP_READ);


private static void handleRead(SelectionKey selectionKey) throws IOException
SocketChannel socketChannel &＃61; (SocketChannel) selectionKey.channel();
ByteBuffer readBuffer &＃61; ByteBuffer.allocate(MESSAGE_LENGTH);
int length &＃61; 0;
while (length < MESSAGE_LENGTH)
length &＃43;&＃61; socketChannel.read(readBuffer);

log.info("receive message from client. client: message length:", socketChannel.getRemoteAddress(), readBuffer.position());
ByteBuffer writeBuffer &＃61; ByteBuffer.allocate(readBuffer.position());
readBuffer.flip();
writeBuffer.put(readBuffer);
writeBuffer.flip();
while (writeBuffer.hasRemaining())
int writeLength &＃61; socketChannel.write(writeBuffer);
log.info("send message to client. client: message length:", socketChannel.getRemoteAddress(), writeLength);





• Debug模式启动服务端NIOSelectorBlockingWriteServer&＃xff0c;在最后一行log.info("send message to client. client ……打上断点
• Debug模式运行上面客户端NIOClient的例子&＃xff0c;向服务端发送100MB的数据&＃xff0c;在ByteBuffer readBuffer &＃61; ByteBuffer.allocate(MESSAGE_LENGTH);打上断点&＃xff0c;运行到断点时&＃xff0c;当服务端调用了socketChannel.write(writeBuffer);向客户端发送的数据还没来得及被客户端接收完时&＃xff08;这时候客户端卡在断点&＃xff0c;并没有开始接收服务端的数据&＃xff09;&＃xff0c;数据会在服务端写缓冲区积压&＃xff0c;在极限情况下&＃xff0c;当服务写端缓冲区写满时&＃xff0c;再调用socketChannel.write(writeBuffer);就写不进去了&＃xff08;返回0&＃xff09;。所以如果在处理完读事件后直接发送数据&＃xff0c;遇到服务端写缓冲区满的情况时&＃xff0c;会直接阻塞当前线程&＃xff08;比如这个例子中会不断执行int writeLength &＃61; socketChannel.write(writeBuffer);并返回0&＃xff09;&＃xff0c;无法及时处理其他客户端请求。
• 这时再起一个线程&＃xff0c;以Debug模式运行NIOClient&＃xff0c;当执行socketChannel.write(writeBuffer);向服务端写数据时&＃xff0c;可能会阻塞&＃xff08;如果发送的数据量太大导致服务端接收缓存写满&＃xff0c;此时服务端线程又在死循环&＃xff0c;所以无法读取客户端发来的数据&＃xff09;&＃xff0c;如果这里不阻塞&＃xff0c;执行到socketChannel.read(readBuffer)就会阻塞&＃xff08;因为服务端一直没有读取到该客户端发送的数据&＃xff0c;肯定也还没执行向客户端发送数据的代码&＃xff09;。这里的客户端是以阻塞模式运行的&＃xff0c;即使把客户端的SocketChannel设置为非阻塞模式&＃xff0c;也是无法及时收到服务端返回的数据的。

所以在传输数据量较大&＃xff0c;需要向客户端回写数据时&＃xff0c;最好注册一个写事件&＃xff0c;避免服务端缓冲区写满时导致线程阻塞&＃xff0c;而无法及时处理其他事件的情况。这样才能体现NIO多路复用模式的特点&＃xff0c;才能可以让一个线程同时为多个客户端服务。

转载请注明出处——胡玉洋 《Java网络编程——NIO处理写事件(SelectionKey.OP_WRITE)》